package com.magupe.im.web.handler.websocket;

import com.alibaba.fastjson.JSON;
import com.magupe.im.web.db.service.DataService;
import com.magupe.im.web.protocol.Command;
import com.magupe.im.web.protocol.CommandPacket;
import com.magupe.im.web.protocol.request.GroupMessageRequest;
import com.magupe.im.web.protocol.request.LoginRequest;
import com.magupe.im.web.protocol.request.MessageRequest;
import com.magupe.im.web.protocol.response.IdleStateResponse;
import com.magupe.im.web.protocol.response.LoginResponse;
import com.magupe.im.web.protocol.response.ResponseStatus;
import com.magupe.im.web.session.Session;
import com.magupe.im.web.session.SessionUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

/**
 * https://www.cnblogs.com/kangniuniu/p/11107768.html
 */
public class WebsocketAccessHandler extends ChannelInboundHandlerAdapter {

    protected Logger logger = LoggerFactory.getLogger(WebsocketAccessHandler.class);

    private WebSocketServerHandshaker handShaker;

    private DataService dataService;

    public WebsocketAccessHandler(DataService dataService){
        this.dataService = dataService;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.debug("websocket connection: unbind session");
        SessionUtils.unBindSession(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 1.传统的HTTP接入，第一次连接，还没有升级得tcp长连接
        if (msg instanceof FullHttpRequest) {
            FullHttpRequest req = (FullHttpRequest) msg;

            processFullHttpRequest(ctx, req);

            // 构造握手响应返回
            String uri = req.uri();
            StringBuilder builder = new StringBuilder().append("ws://").append(req.headers().get(HttpHeaderNames.HOST)).append(uri);
            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(builder.toString(), null, false);
            handShaker = wsFactory.newHandshaker(req);
            if (handShaker == null) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                handShaker.handshake(ctx.channel(), req);
            }

            logger.debug("websocket http request. channelId: {}", ctx.channel().id().toString());
        }

        // 2.webSocket连接开始
        if(msg instanceof CloseWebSocketFrame){
            // 判断是否关闭链路的指令
            handShaker.close(ctx.channel(), ((CloseWebSocketFrame) msg).retain());
        }

        // 3.消息处理
        if(msg instanceof TextWebSocketFrame){
            TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) msg;
            CommandPacket commandPacket = JSON.parseObject(textWebSocketFrame.text(), CommandPacket.class);
            // 1.没有登录
            if(!SessionUtils.hasLogin(ctx.channel())){
                switch (commandPacket.getCommand()){
                    case Command.LOGIN_REQUEST :
                        logger.debug("websocket bind session. channelId: {}", ctx.channel().id().toString());
                        // 绑定session
                        LoginRequest loginRequest = JSON.parseObject(textWebSocketFrame.text(), LoginRequest.class);
                        SessionUtils.bindSession(new Session(loginRequest.getAppId(), loginRequest.getUserId()), ctx.channel());

                        // 返回登录消息
                        ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(new LoginResponse(ResponseStatus.LOGIN_SUCCESS))));

                        // 发送离线消息
                        dataService.sendOfflineMessage(loginRequest.getAppId(), loginRequest.getUserId(), "", ctx.channel());
                        break;
                    default:
                        ReferenceCountUtil.release(msg);
                        ctx.channel().close();
                }
            }
            // 2.登录状态
            switch (commandPacket.getCommand()){
                case Command.MESSAGE_REQUEST :
                    ctx.fireChannelRead(JSON.parseObject(textWebSocketFrame.text(), MessageRequest.class));
                    break;
                case Command.GROUP_MESSAGE_REQUEST :
                    ctx.fireChannelRead(JSON.parseObject(textWebSocketFrame.text(), GroupMessageRequest.class));
                    break;
                case Command.HEART_BEAT_REQUEST :
                    // 收到客户端的心跳包，发送一个回复
                    IdleStateResponse response = new IdleStateResponse(ResponseStatus.IDLE_STATE);
                    logger.debug("websocket heartbeat message: {}", response.toString());
                    ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toString()));
                    break;
                default :
            }
            ReferenceCountUtil.release(msg);
        }
    }

    private void processFullHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req){
        // 1.如果http解码失败，返回http异常
        if (!req.decoderResult().isSuccess()) {
            logger.error("websocket connection: decode error");
            sendHttpResponse(ctx, req);
            return;
        }

        // 2.contextPath is wrong
        String contextPath = req.headers().get("Upgrade");
        if (!"websocket".equals(contextPath)) {
            logger.error("websocket connection: upgrade is wrong");
            sendHttpResponse(ctx, req);
            return;
        }

        // 3.获取http请求url的参数
        String uri = req.uri();
        QueryStringDecoder queryStringDecoder = new QueryStringDecoder(uri);
        Map<String, List<String>> parameters = queryStringDecoder.parameters();

        // 4.没有command参数，不允许通过
        if(null == parameters.get("command") || CollectionUtils.isEmpty(parameters.get("command"))){
            logger.error("websocket connection: illegal connection");
            sendHttpResponse(ctx, req);
            return;
        }

        // 5.command参数不为0,不允许通过
        String command = parameters.get("command").get(0);
        if(StringUtils.isBlank(command) || !"0".equals(command)){
            logger.error("websocket connection: no params connection");
            sendHttpResponse(ctx, req);
            return;
        }
    }

    /**
     * 连接异常处理
     *
     * @param ctx
     * @param req
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req) {
        // 1.返回应答给客户端
        DefaultFullHttpResponse res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NON_AUTHORITATIVE_INFORMATION);
        ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();

        // 2.如果是非Keep-Alive，关闭连接
        ChannelFuture future = ctx.channel().writeAndFlush(res);
        if (!HttpUtil.isKeepAlive(req)) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }
}
